#include "producer/client/MessagePublisher.h"
#include "producer/client/ProducerMediator.h"
#include "codec/RequestBuilder.h"

namespace zeroplus::mq::producer
{

    MessagePublisher::MessagePublisher(std::shared_ptr<ProducerMediator> mediator)
        : mediator_(mediator)
    {
    }

    MessagePublisher::~MessagePublisher()
    {
    }

    std::shared_ptr<ClientNode> MessagePublisher::GetChannelByTopic(std::string topic)
    {
        auto topicRouteInfoTable = getTopicRouteInfoTable();
        routeinfo::TopicRouteInfo topicRouteInfo = topicRouteInfoTable[topic];
        auto [queueData, brokerData, res] = selector_.SelectRouteInfo(topicRouteInfo, ROUND_ROBIN);
        if (!res)
        {
            // LOG_WARN("Select route info failed");
            return nullptr;
        }
        auto brokerAddrIt = brokerData.broker_addrs().find(BrokerState::MASTER);
        assert(brokerAddrIt != brokerData.broker_addrs().end());
        if (clientTable_.find(brokerAddrIt->second) == clientTable_.end())
        {
            // LOG_DEBUG("currentIndex uri=%s", brokerAddrIt->second.c_str());
            Uri uriParser = Uri(brokerAddrIt->second);
            clientTable_[brokerAddrIt->second] = std::make_shared<ClientNode>(uriParser.GetUri());
        }
        return clientTable_[brokerAddrIt->second];
    }

    bool MessagePublisher::Publish(const Header &header, const Bytes &body)
    {
        std::string topic = header.topic;
        auto topicRouteInfoTable = getTopicRouteInfoTable();
        if (topicRouteInfoTable.find(topic) != topicRouteInfoTable.end())
        {
            std::shared_ptr<ClientNode> client = GetChannelByTopic(topic);
            if (!client)
            {
                pendingMessageTable_[topic].enqueue(std::make_pair(header, body));
                return false;
            }
            std::pair<Header, Bytes> pendingMessage;
            while (pendingMessageTable_[topic].try_dequeue(pendingMessage))
            {
                PublishInternal(client, pendingMessage.first, pendingMessage.second);
            }
            return PublishInternal(client, header, body);
        }
        else
        {
            pendingMessageTable_[topic].enqueue(std::make_pair(header, body));
            mediator_->Notify(Publisher, Bytes::from_string(topic));
            return false;
        }
    }

    bool MessagePublisher::PublishInternal(const std::shared_ptr<ClientNode> client, const Header &header, const Bytes &body, int retryCount)
    {
        transport::PublishMessageRequest request;
        request.set_topic(header.topic);
        request.set_queue_id(header.queueId);
        request.set_delay(header.delay);
        request.set_reconsume_times(header.reconsumeTimes);
        request.set_body_type(header.bodyType);
        request.set_body(body.to_string());

        RequestBuilder<transport::RequestCode::PUB_MESSAGE,
                       transport::PublishMessageRequest,
                       transport::PublishMessageResponse>
            builder(client);

        auto [responseCode, response] = builder.InvokeSync(request);

        switch (responseCode)
        {
        case transport::ResponseCode::SUCCESS:
        {
            // LOG_DEBUG("消息发布成功");
            break;
        }
        case transport::ResponseCode::RECV_TIMEOUT:
        {
            if (retryCount < maxRetries_)
            {
                // 实现指数退避等待
                int backoffMs = std::pow(2, retryCount) * initialBackoffMs_;
                std::this_thread::sleep_for(std::chrono::milliseconds(backoffMs));
                LOG_WARN("Message publish failed, retrying... (Attempt: %d)", retryCount + 1);
                return PublishInternal(client, header, body, retryCount + 1);
            }
            else
            {
                LOG_ERROR("Message publish failed after %d retries.", maxRetries_);
                return false;
            }
            break;
        }
        case transport::ResponseCode::UNKNOWN_TYPE:
        {
            // LOG_ERROR("request type %d not supported", requestCmd.code());
            break;
        }
        default:
        {
            break;
        }
        }
        return true;
    }

    void MessagePublisher::InternalReceive(const std::string &topic, const routeinfo::TopicRouteInfo &topicRouteInfo)
    {
        setTopicRouteInfoTable(topic, topicRouteInfo);
    }
}
